// -*-mode:c++; coding:utf-8-*-

#include <boost/asio/ip/tcp.hpp>

#include "logger.hpp"

#include "client_environment.hpp"

namespace xbase
{
	client_environment::client_environment(client_options_ptr options,
					       io_service_pool_ptr pool,
					       const std::string &master_ip,
					       unsigned short master_port)
		: m_options(options)
		, m_io_service_pool(pool)
		, m_master_address(boost::asio::ip::address::from_string(master_ip),
				   master_port)
		, m_stat(options->get_stat_file_name())
	{
		m_io_service_pool->set_options(*m_options);

		m_io_service_pool->run();

		// 启动统计信息计时器
		m_stat_dump_timer.reset(
			new timer_type((m_io_service_pool->get_io_service<io_service_pool::xbase_level>()))
			);

		m_stat_dump_timer->expires_from_now(m_options->get_stat_dump_interval());
		m_stat_dump_timer->async_wait(boost::bind(&this_type::on_stat_dump, this_ptr(), _1));

		// 在service_pool运行后，才可能使用
		m_session_manager.reset(new_async_module(session_manager,
			*m_io_service_pool));
		m_root_meta_cache.reset(new_async_module(root_meta_cache,
			*m_io_service_pool,
			m_master_address,
			*m_session_manager));
		m_table_info_cache.reset(new_async_module(table_info_cache,
			*m_io_service_pool,
			m_master_address,
			*m_session_manager));
		m_location_cache.reset(new_async_module(location_cache,
			*m_io_service_pool,
			*m_root_meta_cache,
			*m_session_manager));
	}

	client_environment::~client_environment()
	{
		m_io_service_pool->stop();
		m_io_service_pool->join();

		// 依次删除
		m_stat_dump_timer.reset();
		m_row_mutator_list.clear();
		m_multi_row_mutator_list.clear();
		m_scanner_list.clear();
		m_parrallel_scanner_list.clear();

		m_location_cache.reset();
		m_table_info_cache.reset();
		m_root_meta_cache.reset();
		m_session_manager.reset();

		m_io_service_pool.reset();
		m_options.reset();
	}

	void client_environment::async_open_table(const std::string &table_name,
		table_info_cache::handler_type handler)
	{
		this->get_table_info_cache().async_get_table_info(table_name, handler);
	}

	row_mutator_ptr client_environment::new_row_mutator()
	{
		return row_mutator_ptr(new_async_module(row_mutator,
			get_io_service_pool(),
			get_session_manager(),
			get_location_cache()));
	}

	multi_row_mutator_ptr client_environment::new_multi_row_mutator()
	{
		return multi_row_mutator_ptr(new_async_module(multi_row_mutator,
			get_io_service_pool(),
			get_session_manager(),
			get_location_cache()));
	}

	parallel_scanner_ptr client_environment::new_parallel_scanner(const scan_info &si,
		parallel_scanner::handler_type handler)
	{
		return parallel_scanner_ptr(new_async_module(parallel_scanner,
			get_io_service_pool(),
			get_session_manager(),
			get_location_cache(),
			si,
			handler));
	}

	scanner_ptr client_environment::new_scanner(const scan_info &si)
	{
		LOG(DEBUG, "scan_info.m_table_name="<<si.m_table_name);
		assert(!si.m_table_name.empty());
		assert(!si.m_column_names.empty());

		scanner_ptr scanner_p(new_async_module(scanner,
			get_io_service_pool(),
			get_session_manager(),
			get_location_cache(),
			si));

		m_io_service_pool->get_io_service<io_service_pool::xbase_level>().dispatch(
			boost::bind(&this_type::on_new_scanner, this_ptr(), scanner_p)
			);

		return scanner_p;
	}

	meta_tablet_scanner_ptr client_environment::new_meta_tablet_scanner(std::string &hostname, uint16_t port,
		std::string &start_row, std::string &end_row)
	{
		LOG(DEBUG, "create meta_tablet_scanner, start_row="<<start_row<<" end_row="<<end_row);
		
		return meta_tablet_scanner_ptr(new_async_module(meta_tablet_scanner,
			get_io_service_pool(),
			boost::asio::ip::tcp::endpoint(
				boost::asio::ip::address::from_string(hostname), port
				),
			get_session_manager(),
			std::string(start_row),
			std::string(end_row)
			));

	}

	void client_environment::async_apply(const row_mutation &mutation,
		row_mutator::handler_type handler)
	{
		xbase::async_apply(get_io_service_pool(),
			get_session_manager(),
			get_location_cache(),
			mutation,
			handler);
	}

	void client_environment::async_apply(const row_mutation &mutation,
		uint64_t timestamp, row_mutator::handler_type handler)
	{
		xbase::async_apply(get_io_service_pool(),
			get_session_manager(),
			get_location_cache(),
			mutation,
			timestamp,
			handler);
	}

	void client_environment::async_apply(const std::string &table_name,
		const multi_row_mutator::container_ptr &mutations, uint64_t timestamp,
		multi_row_mutator::handler_type handler)
	{
		xbase::async_apply(get_io_service_pool(),
			get_session_manager(),
			get_location_cache(),
			table_name,
			mutations,
			timestamp,
			handler);
	}

	void client_environment::async_apply(const std::string &table_name,
		const multi_row_mutator::container_ptr &mutations,
		multi_row_mutator::handler_type handler)
	{
		xbase::async_apply(get_io_service_pool(),
			get_session_manager(),
			get_location_cache(),
			table_name,
			mutations,
			handler);
	}

	void client_environment::async_apply(const table_creation &creation,
		table_mutator::create_handler_type handler)
	{
		xbase::async_apply(get_io_service_pool(),
			get_session_manager(),
			get_master_address(),
			creation,
			handler);
	}

	void client_environment::async_apply(const table_deletion &deletion,
		table_mutator::delete_handler_type handler)
	{
		xbase::async_apply(get_io_service_pool(),
			get_session_manager(),
			get_master_address(),
			deletion,
			handler);
	}

	void client_environment::async_apply(const table_lister &lister,
		table_mutator::list_handler_type handler)
	{
		table_mutator_ptr ptr(new_async_module(table_mutator,
			get_io_service_pool(),
			get_session_manager(),
			get_master_address()));

		ptr->async_apply(lister, handler);
	}

	void client_environment::on_stat_dump(const boost::system::error_code& error)
	{
		//row_mutator_list::iterator rml_iter = m_row_mutator_list.begin();
		//while (rml_iter != m_row_mutator_list.end())
		//{
		//	++rml_iter;
		//}

		//multi_row_mutator_list::iterator mrml_iter = m_multi_row_mutator_list.begin();
		//while (mrml_iter != m_row_mutator_list.end())
		//{
		//	++mrml_iter;
		//}

		scanner_list::iterator sl_iter = m_scanner_list.begin();
		while (sl_iter != m_scanner_list.end())
		{
			scanner_ptr scanner_p = *sl_iter;

			if (scanner_p->is_stopped())
			{
				sl_iter = m_scanner_list.erase(sl_iter);
				continue;
			}

			scanner_p->dump_stat(m_stat.m_scan_stat);

			++sl_iter;
		}

		//parallel_scanner_list::iterator psl_iter = m_parrallel_scanner_list.begin();
		//while (psl != m_parrallel_scanner_list.end())
		//{
		//	++psl_iter;
		//}

		m_stat.write();

		m_stat.reset();
		m_stat_dump_timer->expires_from_now(m_options->get_stat_dump_interval());
		m_stat_dump_timer->async_wait(boost::bind(&this_type::on_stat_dump, this_ptr(), _1));
	}

	void client_environment::on_new_scanner(scanner_ptr scanner_p)
	{
		m_scanner_list.push_back(scanner_p);
	}
}
